package com.simple.datax.conn.impl;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.conn.BaseConnector;
import com.simple.datax.conn.ConnectorPool;
import com.simple.datax.core.model.connpara.CommonConnPara;
import com.simple.datax.service.impl.SendStatusHandler;


/** 
 * @Title: SendConnectorPool.java 
 * @Package com.simple.datax.conn.impl 
 * @Description: This class defines connector pool for sending messages. In fact,
 * this system allows concurrent sending message (not support concurrent receiving).
 * So this class is just an encapsulation on a ConcurrentMap following
 * pool concept.
 * Please refer to interface Sender implementation to check how it support concurrent
 * operation. 
 * @author Joe Jin 
 * @date 2012-12-25 
 * @version V1.0 
 */

public class SendConnectorPool  extends ConnectorPool {

	private static final Logger logger = LoggerFactory
			.getLogger(SendConnectorPool.class);
	private static SendConnectorPool instance;
	private Map<String, BaseConnector> pool;
	private SendStatusHandler handler = null;
	private SendThreadPoolExecutor poolExecutor;
	
//	private SendConnectorPool() {
//		pool = new ConcurrentHashMap<String, BaseConnector>();
//	}
	
	private SendConnectorPool(SendStatusHandler handler, SendThreadPoolExecutor poolExecutor) {
		this.handler = handler;
		this.poolExecutor = poolExecutor;
		pool = new ConcurrentHashMap<String, BaseConnector>();
	}
	
	/**
	 * Singleton
	 * @return SendConnectorPool instance
	 */
	public static synchronized SendConnectorPool getInstance(SendStatusHandler handler, SendThreadPoolExecutor poolExecutor) {
		if (instance == null) {
			instance = new SendConnectorPool(handler, poolExecutor); 
		}
		return instance;
	}
	
	@Override
	public BaseConnector borrowObject(CommonConnPara connPara) {
		String key = getConnKey(connPara.getTransportType(), connPara.getTransportPara());
		BaseConnector connector = null;
		synchronized (pool) {
			connector = pool.get(key);
			
			if (connector == null) {
				ConnectorFactory factory = ConnectorFactory.getInstance();
				try {
					connector = factory.makeObject(key);
					if (handler == null) {
						handler = SendStatusHandler.getInstance();
					} else {
						if (SendStatusHandler.getInstance() != handler) {
							logger.error("Handler created by Spring is different from return by SendStatusHandler.getInstance()");
						}
					}
					connector.setHandler(handler);
					
					if (poolExecutor == null) {
						poolExecutor = SendThreadPoolExecutor.getInstance();
					} else {
						if (SendThreadPoolExecutor.getInstance() != poolExecutor) {
							logger.error("SendThreadPoolExecutor created by Spring is different from return by SendThreadPoolExecutor.getInstance()");
						}
					}
					connector.setPoolExecutor(poolExecutor);
					pool.put(key, connector);
					connector.setEncoding(connPara.getEncoding());
				} catch (Exception e) {
					logger.error("ConnectorFactory makeObject failed.", e);
				}
			}
		}
		return connector;
	}
	
//	@Override
//	public void returnObject(String connType, String connPara,
//			BaseConnector connector) {
//		// Do nothing
//	}
	
	@Override
	public void returnObject(BaseConnector connector) {
		// Do nothing
		
	}
	
}
